Fix send offset to tx, when no active tx #4096
Merged
Conversation
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
moonyoungCHAE
force-pushed
the
issue-4088
branch
9 times, most recently
from
876763a to
6fa0e01
Compare
sobychacko
reviewed
Oct 9, 2025
| } | ||
|
|
||
| private void sendOffsetsToTransaction() { | ||
| if (this.kafkaTxManager != null && TransactionSynchronizationManager.getResource(this.kafkaTxManager.getProducerFactory()) == null) { |
There was a problem hiding this comment.
What if the kafkaTxManager is null? Don't we need to address that also? If it is null, we can return as there is no point in sending the offsets to transaction. Do I miss anything on that?
There was a problem hiding this comment.
@sobychacko You're right. I fixed it. Thanks for the feedback.
. Signed-off-by: moonyougnCHAE <xpf_fl@naver.com>
. Signed-off-by: moonyougnCHAE <xpf_fl@naver.com>
. Signed-off-by: moonyougnCHAE <xpf_fl@naver.com>
moonyoungCHAE
force-pushed
the
issue-4088
branch
from
1e3e9a6 to
a108a33
Compare
sobychacko
reviewed
Oct 10, 2025
| template.send(new ProducerRecord<>(topic11, 0, 0, "bar2")); | ||
| return null; | ||
| }); | ||
| assertThat(successLatch.get().await(30, TimeUnit.SECONDS)).isTrue(); |
There was a problem hiding this comment.
How is this test ensuring that offsets are sent to only active transactions? I mean, i see that you are asserting that two records are consumed transactionally, but how do we ensure the case where there is no active txn? Thanks!
There was a problem hiding this comment.
@sobychacko Thanks for the feedback. I modified the test to verify whether the commit occurred within a transaction by using TransactionExecutionListener.
. Signed-off-by: moonyougnCHAE <xpf_fl@naver.com>
spring-builds
pushed a commit
that referenced
this pull request
Oct 14, 2025
Fixes #4088 Fix `IllegalStateException` when record filtered by interceptor in transactional listener When using a transactional KafkaMessageListenerContainer with an early `RecordInterceptor` that filters out records, the container attempted to send offsets to a transaction even when no active transaction existed. This resulted in an `IllegalStateException` from the Kafka producer: "Cannot send offsets if a transaction is not in progress (currentState=READY)" The issue occurred because: - Container has a `KafkaTransactionManager` configured - Interceptor filters a record before listener execution - No transaction is started (listener never executes) - Container still attempts to send offsets via `sendOffsetsToTransaction()` - Kafka producer rejects the call as no transaction is active Solution: Add a guard clause in `sendOffsetsToTransaction()` to check for an active transaction resource in `TransactionSynchronizationManager`. If either `kafkaTxManager` is null or there's no active transaction resource, the method returns early without attempting to send offsets. The fix includes an integration test in `TransactionalContainerTests` that: - Uses `RecordInterceptor` to filter records - Tracks transaction commits via `TransactionExecutionListener` - Verifies only records that execute listeners trigger transactions - Confirms filtered records don't cause exceptions Signed-off-by: moonyougnCHAE <xpf_fl@naver.com> (cherry picked from commit 81f2941)
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
2 participants
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
Fixes #4088
Description
When a message is filtered out by the Kafka interceptor while the previous message passes through the interceptor and the producer is created, a commit occurs while the Kafka transaction is not active.
structure: Interceptor → Listener (a transactional listener that consumes a message and produces another message)